package cn.itcast.sse;

import cn.itcast.avro.SseAvro;
import cn.itcast.until.DownloadUtil;
import cn.itcast.until.TimeUtil;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.flume.Context;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;

import java.io.*;
import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Flume自定义source
 */
public class SseQuotSource extends AbstractSource implements Configurable, PollableSource {
    private String host;
    private Integer port;
    private String userName;
    private String password;
    private String directory;
    private String fileName;
    private String localDirectory;
    private Integer delay;
    private Integer corePoolSize;
    private Integer maxPoolSize;
    private Long keepAliveTime;
    private Integer capacity;
    ThreadPoolExecutor threadPoolExecutor;

    //业务处理
    @Override
    public Status process() throws EventDeliveryException {
        long time = new Date().getTime();
        if (time > TimeUtil.startTime && time < TimeUtil.endTime) {
            //异步处理
            threadPoolExecutor.execute(new AsyncTask());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return Status.READY;
    }

    //初始化方法
    @Override
    public void configure(Context context) {
        host = context.getString("host");
        port = context.getInteger("port");
        userName = context.getString("userName");
        password = context.getString("password");
        directory = context.getString("ftpDirectory");
        fileName = context.getString("fileName");
        localDirectory = context.getString("localDirectory");
        delay = context.getInteger("delay");
        corePoolSize = context.getInteger("corePoolSize");
        maxPoolSize = context.getInteger("maxPoolSize");
        keepAliveTime = context.getLong("keepAliveTime");
        capacity = context.getInteger("capacity");
        /*
         *初始化线程池
         */
        threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(capacity));
    }

    private class AsyncTask implements Runnable {
        @Override
        public void run() {
            boolean download = DownloadUtil.download(host, port, userName, password, directory, localDirectory);
            try {
                if (download) {
                    InputStreamReader inputStreamReader = new InputStreamReader(new FileInputStream(localDirectory + "/" + fileName));
                    BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
                    String line;
                    int i = 0;
                    while ((line=bufferedReader.readLine())!=null){
                        String[] split = line.split("\\|");
                        if (i==0){
                            //首行判断市场行情状态
                            String s = split[8];
                            if (s.equals("E")){
                                break;
                            }
                        }else {
                            //3.数据转换成avro对象
                            SseAvro sseAvro = transform(split);
                            //4.序列化操作
                            byte[] serialize = serialize(sseAvro);
                            //5.发送管道
                            getChannelProcessor().processEvent(EventBuilder.withBody(serialize));
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    private SseAvro transform(String[] arr) {
        SseAvro sseAvro = new SseAvro();
        sseAvro.setMdStreamID(arr[0].trim());
        sseAvro.setSecurityID(arr[1].trim());
        sseAvro.setSymbol(arr[2].trim());
        sseAvro.setTradeVolume(Long.valueOf(arr[3].trim()));
        sseAvro.setTotalValueTraded(Long.valueOf(arr[4].trim()));
        sseAvro.setPreClosePx(Double.valueOf(arr[5].trim()));
        sseAvro.setOpenPrice(Double.valueOf(arr[6].trim()));
        sseAvro.setHighPrice(Double.valueOf(arr[7].trim()));
        sseAvro.setLowPrice(Double.valueOf(arr[8].trim()));
        sseAvro.setTradePrice(Double.valueOf(arr[9].trim()));
        sseAvro.setClosePx(Double.valueOf(arr[10].trim()));
        sseAvro.setTradingPhaseCode("T11");
        sseAvro.setTimestamp(new Date().getTime());
        return sseAvro;
    }
    private byte[] serialize(SseAvro sseAvro) {

        DatumWriter<SseAvro> specificDatumWriter = new SpecificDatumWriter<>(new SseAvro().getSchema());
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();

        BinaryEncoder binaryEncoder = EncoderFactory.get().directBinaryEncoder(byteArrayOutputStream, null);
        try {
            specificDatumWriter.write(sseAvro, binaryEncoder);
        } catch (IOException e) {
            e.printStackTrace();
        }

        return byteArrayOutputStream.toByteArray();
    }
}
